package com.sleep.ws.server.handler;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.SingleThreadEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;
import java.util.Queue;


/**
 * 监控Netty的任务队列情况，诊断内存泄露的异常情况
 * 对任务队列中的任务数、积压的ByteBuf大小、任务类信息进行监控。
 */
public class MonitorHandler extends ChannelOutboundHandlerAdapter {

    private static final Logger log = LoggerFactory.getLogger(ChannelOutboundHandlerAdapter.class);


    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        //设置高水平位
        ctx.channel().config().setWriteBufferHighWaterMark(1024 * 1024 * 8);
        //设置队列情况监控
        int totalPendingSize = this.monitorPendingTaskCount(ctx);
        long outBoundBufSize = this.monitorOutboundBufSize(ctx);
        //Channel是否可写标识
        boolean writable = ctx.channel().isWritable();
        log.info("chanel[{}] pending task size={}, out buf size={}, writable={}", ctx.channel().remoteAddress(), totalPendingSize, outBoundBufSize, writable);
        //当到达高水位时，Channel的状态被设置为不可写,停止写入数据
        if (ctx.channel().isWritable()) {
            super.write(ctx, msg, promise);
        }
    }

    /**
     * 监控任务队列堆积任务数，任务队列中的任务包括io读写任务，业务程序提交任务
     */
    public int monitorPendingTaskCount(ChannelHandlerContext ctx) {
        int totalPendingSize = 0;
        for (EventExecutor eventExecutor : ctx.executor().parent()) {
            SingleThreadEventExecutor executor = (SingleThreadEventExecutor) eventExecutor;
            // 注意，Netty4.1.29以下版本本pendingTasks()方法存在bug，导致线程阻塞问题
            // 参考 https://github.com/netty/netty/issues/8196
            totalPendingSize += executor.pendingTasks();
        }
//        log.info("任务队列中总任务数 = {} ",totalPendingSize);
        return totalPendingSize;
    }

    /**
     * 监控各个堆积的任务队列中第一个任务的类信息
     */
    public void monitorQueueFirstTask(ChannelHandlerContext ctx) throws NoSuchFieldException, IllegalAccessException {
        Field singleThreadField = SingleThreadEventExecutor.class.getDeclaredField("taskQueue");
        singleThreadField.setAccessible(true);
        for (EventExecutor eventExecutor : ctx.executor().parent()) {
            SingleThreadEventExecutor executor = (SingleThreadEventExecutor) eventExecutor;
            Runnable task = ((Queue<Runnable>) singleThreadField.get(executor)).peek();
            if (null != task) {
                log.info("任务队列中第一个任务信息: {} ", task.getClass().getName());
//                System.out.println("任务队列中第一个任务信息：" + task.getClass().getName());
            }
        }
    }

    /**
     * 监控出站消息的队列积压的byteBuf大小
     */
    public long monitorOutboundBufSize(ChannelHandlerContext ctx) {
        long outBoundBufSize = ((NioSocketChannel) ctx.channel()).unsafe().outboundBuffer().totalPendingWriteBytes();
//        log.info("出站消息队列中积压的buf大小: {} ",outBoundBufSize);
        return outBoundBufSize;
    }

}
